package com.woniu.inlet.subscriber;

import com.rabbitmq.client.Channel;
import com.woniu.outlet.client.ReliableMessageServiceClient;
import com.woniu.outlet.mysql.dao.MessageDao;
import com.woniu.outlet.mysql.dao.SignDao;
import com.woniu.outlet.mysql.po.Message;
import com.woniu.outlet.mysql.po.Sign;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;

import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;

@Slf4j
@Component
@RabbitListener(queues = "user_register_promotion")
public class RegisterMessageSubscriber {
    @Autowired
    private ReliableMessageServiceClient reliableMessageServiceClient;

    @Autowired
    private MessageDao messageDao;

    @Autowired
    private SignDao signDao;

    @Transactional // 抛出异常，事务回滚。
    @RabbitHandler
    public void handle(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) Long tag) throws IOException {
        log.debug("进入监听消息队列，消息为[{}]", message);
        String[] strings = message.split("!");
        if (strings.length < 2) {
            channel.basicReject(tag, false);
            throw new RuntimeException("接收到的消息内容格式错误");
        }
        // 消息的第一部分是：消息在【可靠消息服务】那边的消息表中的ID（唯一）。
        String messageIdString = strings[0];
        Integer messageId = Integer.valueOf(messageIdString);
        /*
         * 消息的第二部分是：新用户的 ID 。
         * 具体的业务逻辑中，这部分内容会有不同。这取决于消息的生产者和消费者之间是如何约定的。
         */
        String userIdString = strings[1];
        Integer userId = Integer.valueOf(userIdString);
        try {
            // 1. 判断是否是重复消息：insert into 去重表
            messageDao.insert(new Message(null, messageId));
        } catch (RuntimeException e) {
            log.debug("该消息重复，我已拒绝签收");
            // 2. 拒绝消息。要求 RabbitMQ 从队列中删除。
            channel.basicReject(tag, false);
            return;
        }
        try {
            // 2. 执行本地业务逻辑（如，给新用户 xxx 发优惠券）
            log.debug("执行本地业务逻辑：为用户[{}]进行签到初始化", userId);
            Sign sign = new Sign();
            sign.setSignDays(0);
            sign.setUserId(userId);
            sign.setStatus(0);
            signDao.insertSelective(sign);
            // 3. 确认接收，并处理该消息：
            channel.basicAck(tag, false);
            // 发起 HTTP 请求: http://可靠消息服务/message/{}/received"
            reliableMessageServiceClient.receivedMessage(messageId);
        } catch (RuntimeException e) {
            // 4. 执行本地业务出错，拒绝消息，要求 RabbitMQ 从队列中删除。
            // 等待发送者的定时器下次再发送一次。
            channel.basicReject(tag, false);
            log.debug("触发 spring-tx 的 @Transactional 的事务回滚，撤销对消息去重表的 insert");
            throw new RuntimeException();
        }
    }
}
